package com.github.zzay.news.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Component;

/**
 * @author zzay
 * @className NewsTaskListener
 * @description News Task Listener
 * @create 2022/06/18 13:14
 */
@Component
public class NewsTaskListener {

    @Autowired
    private ConsumerFactory consumerFactory;

    /**
     * Listen to "editNews" task
     *
     * @param record Record
     */
    @KafkaListener(topics = {"editNews"}, containerFactory = "filterContainerFactory")
    public void onEditNewsTask(ConsumerRecord<?, ?> record) {
        // 消费的哪个 topic、partition 的消息，打印出消息内容、
        System.out.println("简单消费（编辑新闻内容）：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }

    /**
     * Listen to "auditNews" task
     *
     * @param record Record
     */
    @KafkaListener(topics = {"auditNews"}, containerFactory = "filterContainerFactory")
    public void onAuditNewsTask(ConsumerRecord<?, ?> record) {
        // 消费的哪个 topic、partition 的消息，打印出消息内容、
        System.out.println("简单消费（审核新闻内容）：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }

    /**
     * Tasks Filter
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        factory.setAckDiscarded(true);
        factory.setRecordFilterStrategy(consumerRecord -> Integer.parseInt(consumerRecord.value().toString()) % 2 != 0);
        return factory;
    }

}
